package com.zhouchao.register.zookeeper;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.zhouchao.register.RegisterManager;
import com.zhouchao.register.watcher.ServerWatcher;
import com.zhouchao.rpc.common.Server;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @Author: zhouchao
 * @Date: 2021/08/09 16:00
 * @Description:
 */
public class ZookeeperRegisterManager implements RegisterManager {

    public static final Map<String, Server> SERVER_MAP = new ConcurrentHashMap<>();

    private static final String REGISTER_ZOOKEEPER_PREFIX = "/server/register/zookeeper";

    private static CuratorFramework client = null;

    static {
        client = CuratorFrameworkFactory.builder()
                .connectString("192.168.233.101:2181")
                .sessionTimeoutMs(40000)
                .connectionTimeoutMs(50000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.getConnectionStateListenable().addListener((CuratorFramework curatorFramework, ConnectionState connectionState) -> {
            if (ConnectionState.CONNECTED.equals(connectionState)) {
                System.out.println("连接到了注册中心");
            }
        });
        client.start();
        try {
            Stat stat = client.checkExists().forPath(REGISTER_ZOOKEEPER_PREFIX);
            if (stat == null) {
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(REGISTER_ZOOKEEPER_PREFIX);
            }

            client.getData().usingWatcher(new ServerWatcher()).forPath(REGISTER_ZOOKEEPER_PREFIX);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public Boolean register(String ip, int port) {
        Server server = new Server(ip, port);
        String data = JSONUtil.toJsonStr(server);
        String serverPath = ZookeeperRegisterManager.REGISTER_ZOOKEEPER_PREFIX + "/" + ip + ":" + port;
        try {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(serverPath, data.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @Override
    public List<Server> discover() {
        List<Server> serverList = new ArrayList<>();

        String serverPath = ZookeeperRegisterManager.REGISTER_ZOOKEEPER_PREFIX;
        List<String> serverHostList = new ArrayList<>();
        try {
            serverHostList = client.getChildren().forPath(serverPath);
        } catch (Exception e) {
            e.printStackTrace();
        }
        if (CollectionUtil.isNotEmpty(serverHostList)) {
            serverList = serverHostList.stream().map(s -> {
                String nodeValue = getNodeValue(serverPath + "/" + s);
                return JSONUtil.toBean(nodeValue, Server.class);
            }).collect(Collectors.toList());
        }
        return serverList;
    }

    @Override
    public String getNodeValue(String path) {
        String value = null;
        try {
            value = new String(client.getData().forPath(path));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return value;
    }

    @Override
    public boolean updataNodeData(String path, String data) {
        try {
            Stat stat = client.checkExists().forPath(path);
            if (stat == null) {
                return false;
            }
            client.setData().withVersion(stat.getVersion()).forPath(path, data.getBytes()).getVersion();
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public boolean reportServer(String ip, int port, String data) {

        String key = ip + ":" + port;

        String path = ZookeeperRegisterManager.REGISTER_ZOOKEEPER_PREFIX + "/" + key;
        boolean result = updataNodeData(path, data);

        if (result == true) {
            Server server = SERVER_MAP.get(key);
            if (server == null) {
                server = new Server(ip, port);
            }
            if (data != null) {
                server = JSONUtil.toBean(data, Server.class);
            }
            SERVER_MAP.put(key, server);
        }

        return result;
    }

    public boolean reportServer(Server server) {
        String ip = server.getIp();
        int port = server.getPort();
        return reportServer(ip, port, JSONUtil.toJsonStr(server));
    }
}
